package com.dy.framework.config.mqtt;

import com.alibaba.fastjson2.JSON;
import com.dy.common.utils.DateUtils;
import com.dy.common.utils.StringUtils;
import com.dy.common.utils.bean.BeanUtils;
import com.dy.system.domain.*;
import com.dy.system.service.*;
import org.apache.commons.collections4.MapUtils;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallbackExtended;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.io.UnsupportedEncodingException;
import java.util.*;

@Component
public class MqttAcceptCallback implements MqttCallbackExtended {

    private static final Logger logger = LoggerFactory.getLogger(MqttAcceptCallback.class);

    @Autowired
    private IMedicalEquipmentReportLogService medicalEquipmentReportLogService;
    @Autowired
    private IWarningMessageService warningMessageService;
    @Autowired
    private IMedicalInstitutionService medicalInstitutionService;
    @Autowired
    private IFreshMedicalMachineService freshMedicalMachineService;
    @Autowired
    private IFreshWarningMsgTypeService freshWarningMsgTypeService;
    @Autowired
    private IFreshWarningMsgTypeDetailService freshWarningMsgTypeDetailService;



    /**
     * 客户端断开后触发
     *
     * @param throwable
     */
    @Override
    public void connectionLost(Throwable throwable) {
        logger.info("客户端断开后触发 连接断开，可以做重连");
        if (MqttAcceptClient.client == null || !MqttAcceptClient.client.isConnected()) {
            logger.info("emqx重新连接....................................................");
            try {
                MqttAcceptClient.client.reconnect();
                logger.info("--------------------11111111111ClientId:"
                        + MqttAcceptClient.client.getClientId() + "客户端连接成功！--------------------");
                List<FreshMedicalMachine> machineList = freshMedicalMachineService.selectFreshMedicalMachineList(new FreshMedicalMachine());
                logger.error("machineList = {}",machineList.size());
//                for (FreshMedicalMachine medicalMachine : machineList) {
//                    String topicName = medicalMachine.getMachineCode()+"12345678/pub_data";
//                    logger.info("subscribe topicName ={}",topicName);
//                    MqttAcceptClient.client.subscribe(topicName,2);
//                }
                MqttAcceptClient.getClient().subscribe("294147031FFD0509612345678/pub_data",2);
                MqttAcceptClient.getClient().subscribe("415048127CAAC320112345678/pub_data",2);

                MqttAcceptClient.getClient().subscribe("41513908539F0220812345678/pub_data",2);

                MqttAcceptClient.getClient().subscribe("28203016535FE529012345678/pub_data",2);

            } catch (MqttException e) {
                logger.error("订阅主题失败 error ={}",e.getMessage());
            }
        }
    }

    /**
     * 客户端收到消息触发
     *
     * @param topic       主题
     * @param mqttMessage 消息
     */
    @Override
    public void messageArrived(String topic, MqttMessage mqttMessage) throws Exception {
        logger.info("MqttAcceptCallback 接收消息主题 : " + topic);
        logger.info("MqttAcceptCallback 接收消息Qos : " + mqttMessage.getQos());
        logger.info("MqttAcceptCallback 接收消息内容 : " + new String(mqttMessage.getPayload()));
        //488098178EE9C527012345678/pub_data
        if (StringUtils.isBlank(topic) || StringUtils.isBlank(new String(mqttMessage.getPayload()))) {
            return;
        }
        String code = topic.substring(0, 17);
        String topicName = topic.substring(topic.indexOf("/") + 1);
        if ("pub_data".equals(topicName)) {
            savePubData(code, topicName, new String(mqttMessage.getPayload()));
        }
    }

    public static void main(String[] args) {
        String topic = "488098178EE9C527012345678/pub_data";
        String code = topic.substring(0, 17);
        String topicName = topic.substring(topic.indexOf("/") + 1);
        String substring = topic.substring(17, topic.indexOf("/"));
        System.out.println(code);
        System.out.println(topicName);
        System.out.println(substring);
    }

    private void savePubData(String code, String topicName, String mqttMessage) {
        MedicalEquipmentReportLog reportLog = new MedicalEquipmentReportLog();
        FreshMedicalMachine freshMedicalMachine = freshMedicalMachineService.selectMedicalMachineByCode(code);
        if (freshMedicalMachine == null) {
            logger.error("设备编码code = {} 没绑定，上报的信息不作存储",code);
            return;
        }
        Map<String, Object> jsonMap = JSON.parseObject(mqttMessage, HashMap.class);
        Object variant = MapUtils.getObject(jsonMap, "Variant");
        if (variant instanceof List) {
            ArrayList variantList = (ArrayList) variant;
            for (Object obj : variantList) {
                if (obj instanceof Map) {
                    //设备数据上报
                    reportLog.setCode(code);
                    Map<String, Object> objectMap = (Map) obj;
                    String unix = MapUtils.getString(objectMap, "Unix");
                    reportLog.setBaojingtime(unix);
                    Map<String, Object> pubDataMap = (Map<String, Object>) MapUtils.getMap(objectMap, "Pub_Data");
                    Map<String, Object> rxdqMap = (Map<String, Object>) MapUtils.getMap(pubDataMap, "RXDQ");
                    Map<String, Object> LocalghostMap = (Map<String, Object>) MapUtils.getMap(pubDataMap, "Localghost");
                    String onoff = MapUtils.getString(rxdqMap, "onoff");
                    reportLog.setOnoff(onoff);
                    String sewageL = MapUtils.getString(rxdqMap, "sewageL");
                    reportLog.setSewagel(sewageL);
                    String sewageT = MapUtils.getString(rxdqMap, "sewageT");
                    reportLog.setSewaget(sewageT);
                    String inPumpOnoff = MapUtils.getString(rxdqMap, "inPumpOnoff");
                    reportLog.setInpumponoff(inPumpOnoff);
                    String outPumpOnoff = MapUtils.getString(rxdqMap, "outPumpOnoff");
                    reportLog.setOutpumponoff(outPumpOnoff);
                    String ozoneOnoff = MapUtils.getString(rxdqMap, "ozoneOnoff");
                    reportLog.setOzoneonoff(ozoneOnoff);
                    String sterilizeTime = MapUtils.getString(rxdqMap, "cirPumpOnoff");
                    reportLog.setSterilizetime(sterilizeTime);
                    String clearwaterLow = MapUtils.getString(rxdqMap, "clearwaterLow");
                    reportLog.setClearwaterlow(clearwaterLow);
                    String clearwaterHigh = MapUtils.getString(rxdqMap, "clearwaterHigh");
                    reportLog.setClearwaterhigh(clearwaterHigh);
                    String onoffwrite = MapUtils.getString(rxdqMap, "onoffwrite");
                    reportLog.setOnoffwrite(onoffwrite);
                    String xiwubengbaojing = MapUtils.getString(rxdqMap, "xiwubengbaojing");
                    reportLog.setXiwubengbaojing(xiwubengbaojing);
                    String paiwubengbaojing = MapUtils.getString(rxdqMap, "paiwubengbaojing");
                    reportLog.setPaiwubengbaojing(paiwubengbaojing);
                    String longitude = MapUtils.getString(LocalghostMap, "GPS经度-高德");
                    reportLog.setLongitude(longitude);
                    String dimension = MapUtils.getString(LocalghostMap, "GPS纬度-高德");
                    reportLog.setDimension(dimension);
                    String runTime = MapUtils.getString(LocalghostMap, "设备运行时间");
                    String networkStatus = MapUtils.getString(LocalghostMap, "4G信号强度");

                    reportLog.setRuntime(runTime);
                    reportLog.setCreateTime(DateUtils.parseDateToStr(DateUtils.YYYY_MM_DD,new Date()));
                    reportLog.setCrTime(new Date());

                    //报警信息
                    WarningMessage message = new WarningMessage();
                    MedicalInstitution institution = medicalInstitutionService.selectMedicalInstitutionById(freshMedicalMachine.getInstitutionId());
                    if (institution != null) {
                        message.setMedicalId(institution.getId());
                        reportLog.setMedicalId(institution.getId());
                    }
                    /***
                     * 设备状态告警信息
                     */
                    message.setCode(code);
                    message.setOnoff(onoff);
                    message.setClearwaterlow(clearwaterLow);
                    message.setInpumponoff(inPumpOnoff);
                    message.setOutpumponoff(outPumpOnoff);
                    message.setOzoneonoff(ozoneOnoff);
                    message.setSterilizetime(sterilizeTime);
                    message.setXiwubengbaojing(xiwubengbaojing);
                    message.setPaiwubengbaojing(paiwubengbaojing);
                    message.setBaojingtime(unix);
                    message.setCreateTime(new Date());
                    warningMessageService.insertWarningMessage(message);



                    if (StringUtils.isNotBlank(networkStatus)){
                        logger.info("networkStatus = {} and code = {}",networkStatus,code);
                        String warningMsg = "0".equals(networkStatus) ? "网络状态异常" : "网络恢复正常";
                        Integer gjType = "0".equals(networkStatus) ? 1 : 2;
                        Integer warningLevel = 0;
                        //不正常
                        if ("0".equals(networkStatus)){
                            FreshWarningMsgTypeDetail detail = freshWarningMsgTypeDetailService.selectWarningTypeByType("networkStatus");
                            //第一条是恢复，那么告警级别直接是低
                            if (detail == null || detail.getGjType() == 2){
                                warningLevel = 1;
                            }else if (detail != null && detail.getGjType() == 1){
                                long millisecond = DateUtils.differentByMillisecond(new Date(), detail.getCreateTime());
                                if ( 5 < millisecond && millisecond < 1440){
                                    warningLevel = 1;
                                }else if (millisecond > 1440 && millisecond < 2880){
                                    warningLevel = 2;
                                }else {
                                    warningLevel = 3;
                                }
                            }
                        }
                        saveWarningMsgType("networkStatus",warningMsg,code,institution.getId(),gjType,warningLevel,onoff);
                    }

                    if (StringUtils.isNotBlank(clearwaterLow)){
                        String warningMsg = "0".equals(clearwaterLow) ? "纯净水位低" : "纯净水位正常";
                        Integer gjType = "0".equals(clearwaterLow) ? 1 : 2;
                        Integer warningLevel = "0".equals(clearwaterLow) ? 3 : 0;
                        saveWarningMsgType("clearwaterLow",warningMsg,code,institution.getId(),gjType,warningLevel,onoff);
                    }

                    if (StringUtils.isNotBlank(xiwubengbaojing)){
                        String warningMsg = "1".equals(xiwubengbaojing) ? "吸污泵超时报警" : "吸污泵运行正常";
                        Integer gjType = "1".equals(xiwubengbaojing) ? 1 : 2;
                        Integer warningLevel = "1".equals(xiwubengbaojing) ? 3 : 0;
                        saveWarningMsgType("chaoshibaojing",warningMsg,code,institution.getId(),gjType,warningLevel,onoff);
                    }

                    if (StringUtils.isNotBlank(paiwubengbaojing)){
                        String warningMsg = "1".equals(paiwubengbaojing) ? "排污泵超时报警" : "排污泵运行正常";
                        Integer gjType = "1".equals(paiwubengbaojing) ? 1 : 2;
                        Integer warningLevel = "1".equals(paiwubengbaojing) ? 3 : 0;
                        saveWarningMsgType("chaoshibaojing",warningMsg,code,institution.getId(),gjType,warningLevel,onoff);
                    }

                    medicalEquipmentReportLogService.insertMedicalEquipmentReportLog(reportLog);
                    //更新设备信息数据
                    FreshMedicalMachine medicalMachine = BeanUtils.generatorObject(reportLog, FreshMedicalMachine.class);
                    medicalMachine.setMachineCod(code);
                    medicalMachine.setId(freshMedicalMachine.getId());
                    medicalMachine.setInstitutionId(institution.getId());
                    medicalMachine.setNetworkStatus("0".equals(networkStatus) ? "0" : "1");
                    medicalMachine.setMachinePh(null);
                    medicalMachine.setMachineCod(null);
                    medicalMachine.setMachineAy(null);
                    medicalMachine.setRunningStatus(onoff);
                    freshMedicalMachineService.updateFreshMedicalMachine(medicalMachine);

                }
            }
        }
    }



    public void saveWarningMsgType(String warningType,String warningMsg,String code,Long medicalId,Integer gjType,Integer warningLevel,String onoff){
        FreshWarningMsgTypeDetail msgType = new FreshWarningMsgTypeDetail();
        msgType.setWarningType(warningType);
        msgType.setCode(code);
        msgType.setMedicalId(medicalId);
        msgType.setWarningMsg(warningMsg);
        msgType.setGjType(gjType);
        msgType.setWarningLevel(warningLevel);
        msgType.setRunningStatus(onoff);
        freshWarningMsgTypeDetailService.insertFreshWarningMsgTypeDetail(msgType);
    }


    /**
     * 发布消息成功
     *
     * @param token token
     */
    @Override
    public void deliveryComplete(IMqttDeliveryToken token) {
        String[] topics = token.getTopics();
        for (String topic : topics) {
            logger.info("MqttAcceptCallback 向主题：" + topic + "发送消息成功！");
        }
        try {
            MqttMessage message = token.getMessage();
            byte[] payload = message.getPayload();
            String s = new String(payload, "UTF-8");
            logger.info("MqttAcceptCallback 消息的内容是：" + s);
        } catch (Exception e) {
            logger.error("发送消息失败");
        }
    }

    /**
     * 连接emq服务器后触发
     *
     * @param b
     * @param s
     */
    @Override
    public void connectComplete(boolean b, String s) {
        logger.info("--------------------ClientId:"
                + MqttAcceptClient.client.getClientId() + "客户端连接成功！--------------------");
        // 以/#结尾表示订阅所有以test开头的主题
        try {
            MqttAcceptClient.client.subscribe("409124023F4CA189812345678/pub_data", 2);
        } catch (MqttException e) {
           logger.error("链接失败");
        }
    }
}
